package kafka

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

type WorkerPool struct {
	Config
	Jobs chan Job
	Workers chan Worker
	parent context.Context
	mu sync.Mutex
	WorkerCount int
}

type Config struct {
	time.Duration
	Max int
	Idle int
}

func NewConfig() Config {
	return Config{
		Duration: time.Second*10,
		Max:      10,
		Idle:     5,
	}
}

type Job struct {
	Work func()
}

type Worker struct {

}

func (w *WorkerPool) NewWorkder() *Worker {
	return &Worker{}
}

func NewWorkerPool(conf Config) *WorkerPool {
	return &WorkerPool{
		Jobs: make(chan Job),
		parent: context.Background(),
		WorkerCount: conf.Idle,
		Config:conf,
	}
}

func (w *WorkerPool) AddJob(job Job)  {

	go func() {
		w.mu.Lock()
		defer w.mu.Unlock()
		w.Jobs<-job
	}()
	fmt.Println(len(w.Jobs),"----------1")
	if w.WorkerCount<w.Max&&len(w.Jobs)>0{
		w.WorkerCount++
		go func() {
			ctx,cancel:=context.WithCancel(w.parent)
			quit:=make(chan os.Signal)
			signal.Notify(quit, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT,os.Kill,os.Interrupt)

			for {
				select {
				case <-ctx.Done():
					log.Fatal("context canceled.")
				case <-quit:
					log.Fatal("received quit.")
					cancel()
				case <-time.After(w.Duration):
					log.Fatal("timeout and release this worker.")
					w.WorkerCount--
					return
				case job:=<-w.Jobs:
					fmt.Println("ex go")
					job.Work()
				}
			}
		}()
	}
}


func (w *WorkerPool) Start()  {
	for i:=0;i<w.Idle;i++{
		go func() {
			ctx,cancel:=context.WithCancel(w.parent)
			quit:=make(chan os.Signal)
			signal.Notify(quit, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT,os.Kill,os.Interrupt)
			for {
				select {
				case <-ctx.Done():
					log.Fatal("context canceled.")
				case <-quit:
					log.Fatal("received quit.")
					cancel()
				case job:=<-w.Jobs:
					time.Sleep(5*time.Second)
					job.Work()
				}
			}
		}()
	}
}

